package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime220815.bean.TradeProvinceOrderBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/19 16:15
 **/

public class DwsTradeOrderProvinceWindow_r {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_trade_order_detail_r";
        String groupId = "DwsTradeOrderProvinceWindow_r_group";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(flinkKafkaConsumer);
        //{"id":"31841","order_id":"29530","user_id":"100","course_id":"406","course_name":"大数据技术之Hive（2020版）",
        // "final_amount":"200.0","province_id":"3","date_id":"2023-02-19","create_time":"2023-02-19 16:14:37","ts":"1676794476"}

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

        SingleOutputStreamOperator<TradeProvinceOrderBean> beanDS = jsonObjDS.map(
                new MapFunction<JSONObject, TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean map(JSONObject value) throws Exception {
                        String provinceId = value.getString("province_id");
                        Double final_amount = value.getDouble("final_amount");
                        String user_id = value.getString("user_id");
                        Long ts = value.getLong("ts");

                        TradeProvinceOrderBean bean = TradeProvinceOrderBean.builder()
                                .provinceId(provinceId)
                                .amount(final_amount)
                                .count_order(1L)
                                .setUid(new HashSet<>(Collections.singleton(user_id)))
                                .ts(ts)
                                .build();
                        return bean;
                    }
                }
        );

        SingleOutputStreamOperator<TradeProvinceOrderBean> withWatermark = beanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
                    @Override
                    public long extractTimestamp(TradeProvinceOrderBean element, long recordTimestamp) {
                        return element.getTs()*1000;
                    }
                }));

        KeyedStream<TradeProvinceOrderBean, String> keyDS = withWatermark.keyBy(value -> value.getProvinceId());

        WindowedStream<TradeProvinceOrderBean, String, TimeWindow> windowDS = keyDS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                        value1.getSetUid().addAll(value2.getSetUid());
                        value1.setCount_order(value1.getCount_order() + value2.getCount_order());
                        value1.setAmount(value1.getAmount() + value2.getAmount());
                        return value1;
                    }
                },
                new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderBean> input, Collector<TradeProvinceOrderBean> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        long now = System.currentTimeMillis();

                        for (TradeProvinceOrderBean tradeProvinceOrderBean : input) {
                            tradeProvinceOrderBean.setStt(DateFormatUtil.toYmdHms(start));
                            tradeProvinceOrderBean.setEdt(DateFormatUtil.toYmdHms(end));
                            tradeProvinceOrderBean.setTs(now);
                            tradeProvinceOrderBean.setCount_people((long) tradeProvinceOrderBean.getSetUid().size());
                            out.collect(tradeProvinceOrderBean);
                        }

                    }
                }
        );

        SingleOutputStreamOperator<TradeProvinceOrderBean> withProvinceNameDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeProvinceOrderBean>("dim_base_province") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, TradeProvinceOrderBean obj) {
                        String name = dimInfoJsonObj.getString("name".toUpperCase());
                        obj.setProvinceName(name);
                    }

                    @Override
                    public String getKey(TradeProvinceOrderBean obj) {
                        return obj.getProvinceId();
                    }
                }, 60, TimeUnit.SECONDS

        );

        withProvinceNameDS.addSink(ClickHouseUtils.getClickHouseSinkFunction(
                "insert into dws_trade_order_province values(?,?,?,?,?,?,?,?)"
        ));

        env.execute();
    }
}
